package com.ztrue.framework.disruptor;

import com.ztrue.framework.Config;
import com.ztrue.framework.context.ICommandTransmit;
import com.ztrue.framework.core.command.CommandMessage;
import com.ztrue.framework.core.command.ICommand;
import com.ztrue.framework.disruptor.event.CommandMessageFactory;
import com.ztrue.framework.exception.NotFoundException;
import com.ztrue.common.log.Logger;
import com.ztrue.common.properties.Prop;
import com.ztrue.common.properties.Proper;

import java.util.HashMap;
import java.util.Map;

/**
 * 使用disruptor作为命令处理总线
 *
 * @author zhangw
 * @date 2017/1/18 16:21
 */
public class DisruptorCommandBus implements ICommandTransmit {
    private static final Logger logger = Logger.getLogger(DisruptorCommandBus.class);
    private Map<String, IDistributeMessageQueue<CommandMessage>> commandQueueMap = new HashMap<>();
    private IDistributeMessageQueue<CommandMessage> defaultCommandQueue;

    public DisruptorCommandBus(MessageHandler<CommandMessage> messageHandler) {
        Prop prop = Proper.getProp(Config.CONFIG_PATH);
        if (prop == null) {
            logger.error("无法找到基础配置文件");
            throw new NotFoundException("无法找到基础配置文件");
        }
        int poolSize = prop.getInt("command.queue.pool.size", 4);
        int bufferSize = Proper.getInt("command.queue.pool.buffer.size", 1024);
        //int threadNumbers = Proper.getInt("command.queue.pool.thread.numbers", 2);
        defaultCommandQueue = new MessageQueuePool<>(poolSize, bufferSize, 2, messageHandler, new CommandMessageFactory());
        commandQueueMap.put("default", defaultCommandQueue);
    }

    /**
     * 执行一个命令
     *
     * @param command 命令消息
     */
    @Override
    public void send(ICommand command) {
        if (commandQueueMap.containsKey(command.getClass().getName())) {
            //// TODO: 定制专属通道
        } else {
            defaultCommandQueue.send(cm -> cm.setMessage(command), command.getRoutingKey());
        }
    }
}